
#pragma once

#include "mosquitto.h"
#include <atomic>
#include <condition_variable>
#include <functional>
#include <iostream>
#include <mutex>
#include <queue>
#include <string>
#include <vector>

#include "common.hpp"

class Msg_t {
public:
    std::string topic;
    std::string payload;

public:
    Msg_t() = delete;
    Msg_t(Msg_t &&) = default;

    Msg_t(const std::string &t, const std::string &p)
        : topic(t)
        , payload(p)
    {
    }

    Msg_t(const Msg_t &) = default;

    Msg_t &operator=(const Msg_t &) = default;

    ~Msg_t() = default;
};

class MqttClient {
public:
    std::vector<std::function<bool()>> doConnected; /* 在连接的时候要做的事情 */
    std::string _clientId;                          /* 客户端标识 */
private:
    std::atomic<bool> _isConnected; /* 是否连接标志 */
    struct mosquitto *_mosquitto;   /* 句柄 */
public:
    /* 消息队列 */
    std::queue<Msg_t> _mqttMsgQueue;
    std::mutex _mutex;
    std::condition_variable _cv;

private:
    static void onConnect(struct mosquitto *mosq, void *userData, int result);
    static void onDisconnect(struct mosquitto *mosq, void *userData, int result);
    static void onMessage(struct mosquitto *mosq, void *userData, const mosquitto_message *message);

public:
    MqttClient(const std::string &clientId);

    MqttClient() = delete;                   /* 不允许无参构造 */
    MqttClient(const MqttClient &) = delete; /* 不允许拷贝构造 */
    MqttClient &operator=(const MqttClient &) = delete;
    MqttClient(MqttClient &&) = delete; /* 不允许移动拷贝 */

    bool connect(const std::string &host, int port = 1883, int keepAlive = 60);
    bool connect(const std::string &host, const std::string &userName, const std::string &passwd, int port = 1883, int keepAlive = 60);

    bool disconnect(void);

    bool publish(const std::string &topic, const std::string &payload);

    bool subscribe(const std::string &topic);

    bool unSubscribe(const std::string &topic);

    bool isConnected(void);

    void registerOnConnected(const std::function<bool()> &func);

    Msg_t getOneItem(void);

    ~MqttClient();
};

MqttClient::MqttClient(const std::string &clientId)
    : _clientId(clientId)
    , _isConnected(false)
    , _mosquitto(nullptr)
{
    mosquitto_lib_init();

    _mosquitto = mosquitto_new(_clientId.c_str(), true, this);
    if (!_mosquitto) {
        throw std::runtime_error("Failed to create Mosquitto instance");
    }

    mosquitto_connect_callback_set(_mosquitto, onConnect);
    mosquitto_disconnect_callback_set(_mosquitto, onDisconnect);
    mosquitto_message_callback_set(_mosquitto, onMessage);
}

bool MqttClient::connect(const std::string &host, int port, int keepAlive)
{
    if (_isConnected) {
        return true;
    }

    mosquitto_connect(_mosquitto, host.c_str(), port, keepAlive);

    return !mosquitto_loop_start(_mosquitto);
}

bool MqttClient::connect(const std::string &host, const std::string &userName, const std::string &passwd, int port, int keepAlive)
{
    if (_isConnected) {
        return true;
    }

    mosquitto_username_pw_set(_mosquitto, userName.c_str(), passwd.c_str());

    mosquitto_connect(_mosquitto, host.c_str(), port, keepAlive);

    return !mosquitto_loop_start(_mosquitto);
}

bool MqttClient::disconnect(void)
{
    if (!_isConnected) {
        return true;
    }

    mosquitto_disconnect(_mosquitto);
    return !mosquitto_loop_stop(_mosquitto, true);
}

bool MqttClient::publish(const std::string &topic, const std::string &payload)
{
    mqtt_info("pubish topic --> {}", topic);
    mqtt_info("publish payload --> {}", payload);

    if (!_isConnected) {
        mqtt_error("mqtt not connect, publish failed, please check mqtt connect param or mqtt server");
        return true;
    }

    return !mosquitto_publish(_mosquitto, nullptr, topic.c_str(), payload.length(), payload.c_str(), 0, false);
}

bool MqttClient::subscribe(const std::string &topic)
{
    if (!_isConnected) {
        return true;
    }

    mqtt_info("subscribe topic --> {}", topic);

    return !mosquitto_subscribe(_mosquitto, nullptr, topic.c_str(), 0);
}

bool MqttClient::unSubscribe(const std::string &topic)
{
    mqtt_info("unSubscribe topic --> {}", topic);
    return !mosquitto_unsubscribe(_mosquitto, nullptr, topic.c_str());
}

void MqttClient::onConnect(struct mosquitto *mosq, void *userData, int result)
{
    MqttClient *client = reinterpret_cast<MqttClient *>(userData);
    if (result == MOSQ_ERR_SUCCESS) {
        client->_isConnected = true;

        mqtt_info("{} connect success", client->_clientId);

        std::for_each(client->doConnected.begin(), client->doConnected.end(), [](const std::function<bool()> &f) {if (f) {f();} });

    } else {
        mqtt_error("Failed to connect to MQTT broker: {}", mosquitto_connack_string(result));
    }
}

void MqttClient::onDisconnect(struct mosquitto *mosq, void *userData, int result)
{
    MqttClient *client = reinterpret_cast<MqttClient *>(userData);
    client->_isConnected = false;
}

void MqttClient::onMessage(struct mosquitto *mosq, void *userData, const mosquitto_message *message)
{
    MqttClient *client = reinterpret_cast<MqttClient *>(userData);

    mqtt_info("{} recv topic <-- {}", client->_clientId, static_cast<char *>(message->topic));
    mqtt_info("{} recv payload <-- {}", client->_clientId, static_cast<char *>(message->payload));

    {
        std::unique_lock<std::mutex> _lock(client->_mutex);

        client->_mqttMsgQueue.emplace(Msg_t(static_cast<char *>(message->topic), static_cast<char *>(message->payload)));

        client->_cv.notify_one();
    }
}

Msg_t MqttClient::getOneItem(void)
{
    std::unique_lock<std::mutex> _lock(_mutex);

    _cv.wait(_lock, [this] { return !this->_mqttMsgQueue.empty(); });

    Msg_t msg = _mqttMsgQueue.front();

    _mqttMsgQueue.pop();

    return msg;
}

void MqttClient::registerOnConnected(const std::function<bool()> &func)
{
    doConnected.emplace_back(func);
}

bool MqttClient::isConnected(void)
{
    return _isConnected;
}

MqttClient::~MqttClient()
{
    if (_mosquitto) {
        mosquitto_destroy(_mosquitto);
        mosquitto_lib_cleanup();
    }
}